/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.servicemix.audit.async;

import java.io.Serializable;

import javax.jbi.messaging.MessageExchange;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.servicemix.jbi.event.ExchangeEvent;
import org.apache.servicemix.jbi.event.ExchangeListener;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.jms.listener.DefaultMessageListenerContainer;

public abstract class AbstractJmsExchangeListener<T> implements ExchangeListener, InitializingBean {

    private static final String EVENT = "ExchangeListenerEvent";

    private enum Event {
        Sent, Accepted
    }

    private ConnectionFactory factory;
    private DefaultMessageListenerContainer container;
    private JmsTemplate jmsTemplate;

    public void exchangeAccepted(final ExchangeEvent event) {
        jmsTemplate.send(new MessageCreator() {

            public Message createMessage(Session session) throws JMSException {
                createConfigNodeForNamespace(event.getExchange());
                ObjectMessage message = session.createObjectMessage();
                message.setStringProperty(EVENT, Event.Accepted.name());
                message.setObject((Serializable) getSerializer(event.getExchange().getService().getNamespaceURI()).serialize(
                        event.getExchange()));
                return message;
            }

        });
    }

    public void exchangeSent(final ExchangeEvent event) {
        jmsTemplate.send(new MessageCreator() {

            public Message createMessage(Session session) throws JMSException {
                createConfigNodeForNamespace(event.getExchange());
                ObjectMessage message = session.createObjectMessage();
                message.setObjectProperty(EVENT, Event.Sent.name());
                message.setObject((Serializable) getSerializer(event.getExchange().getService().getNamespaceURI()).serialize(
                        event.getExchange()));
                return message;
            }

        });
    }

    public void afterPropertiesSet() throws Exception {
        if (factory == null) {
            throw new IllegalStateException("Unable to initialize JMS ExchangeListener -- no ConnectionFactory set");
        }
        container = new DefaultMessageListenerContainer();
        container.setConnectionFactory(factory);
        container.setDestinationName(getDestinationName());
        container.setMessageListener(new MessageListener() {

            @SuppressWarnings("unchecked")
            public void onMessage(Message message) {
                ObjectMessage om = (ObjectMessage) message;
                try {
                    T object = (T) om.getObject();
                    switch (Event.valueOf(om.getStringProperty(EVENT))) {
                    case Sent:
                        onSent(object);
                    case Accepted:
                        onAccepted(object);
                    }
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }

        });
        container.afterPropertiesSet();
        container.setConcurrentConsumers(10);
        container.start();

        jmsTemplate = new JmsTemplate(factory);
        jmsTemplate.setDefaultDestinationName(getDestinationName());
    }

    protected String getDestinationName() {
        return getClass().getName();
    }

    public void setConnectionFactory(ConnectionFactory factory) {
        this.factory = factory;
    }

    public abstract void onAccepted(T accepted);

    public abstract void onSent(T sent);

    protected abstract ExchangeSerializer<T> getSerializer(String namespace);
    
    /**
     * 
     * Create node for configure of namespace
     * 
     * @param exchange
     * 
     */
    protected abstract void createConfigNodeForNamespace(MessageExchange exchange);

}
